package com.hong.qrpc.client;

import com.hong.qrpc.Peer;
import com.hong.qrpc.TransportClient;
import com.hong.qrpc.constant.QrpcConst;
import com.hong.qrpc.exception.transport.DisconnectException;
import com.hong.qrpc.factory.ResolverFactory;
import com.hong.qrpc.handler.factory.HandlerFactory;
import com.hong.qrpc.handler.in.listener.CloseListener;
import com.hong.qrpc.header.QrpcHeader;
import com.hong.qrpc.resolver.Resolver;
import com.hong.qrpc.task.NettyTask;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @Author HQT
 * @Date 2022/8/8 20:30
 */
@Slf4j
@Data
public class NettyTransportClient implements TransportClient {
    private EventLoopGroup workGroup;
    private Channel channel;
    private int protocolType;
    private Peer helpCompare;
    private volatile boolean closed = false;
    /**
     * 管理请求的map     requestId ->
     */
    private ConcurrentHashMap<Long, NettyTask<InputStream>> resultMap = new ConcurrentHashMap();
    private AtomicLong requestIdFactory = new AtomicLong();

    @Override
    public void connect(Peer peer) {
        NettyTransportClient thisClient = this;
        helpCompare = peer;
        workGroup = new NioEventLoopGroup();
        this.protocolType = peer.getProtocolType();
        try {
            Channel channel = new Bootstrap()
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator())
                    .group(workGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel nsc) throws Exception {
                            nsc.pipeline().addLast(HandlerFactory.getResponseHandler(protocolType, resultMap))
                                    .addLast(new CloseListener(thisClient));
                        }
                    })
                    .connect(new InetSocketAddress(peer.getHost(), peer.getPort()))
                    .sync()
                    .channel();
            this.channel = channel;
        } catch (Exception e) {
            workGroup.shutdownGracefully();
            throw new RuntimeException(e);
        } finally {
            //一定要放锁
            synchronized (peer) {
                peer.notifyAll();
            }
        }
    }

    /**
     * 因为单次写入是线程安全的，所以该方法在为数据添加完请求头之后就可以发送消息了
     *
     * @param data              写入的数据
     * @param serializationType
     * @param timeout
     * @return
     */
    @Override
    public InputStream write(InputStream data, int serializationType, long timeout) {
        //说明连接的时候出现异常了，需要踢出该客户端
        if (channel == null || closed) {
            throw new DisconnectException();
        }
        long requestId = requestIdFactory.getAndIncrement();
        NettyTask<InputStream> nettyTask = new NettyTask<>();
        resultMap.put(requestId, nettyTask);
        //如果超时时间小于一的话就默认超时时间为无限长
        if (timeout < 1) {
            timeout = Integer.MAX_VALUE;
        }
        try {
            //add request header and request body
            QrpcHeader qrpcHeader = new QrpcHeader().setSerializationType(serializationType).setDataLen(data.available())
                    .setVersion(QrpcConst.QRPC_VERSION).setRequestId(requestId);
            Resolver resolver = ResolverFactory.getResolver(protocolType);
            byte[] header = resolver.toByteArr(qrpcHeader);
            ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(header.length + data.available());
            buffer.writeBytes(header);
            buffer.writeBytes(data, data.available());
            //send request
            log.debug("send request  request header:{}", qrpcHeader);
            channel.writeAndFlush(buffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            //get result
            return nettyTask.get(timeout);
        } catch (Exception e) {
            throw new IllegalArgumentException("service timeout");
        }
    }

    @Override
    public void close() {
        workGroup.shutdownGracefully();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        NettyTransportClient that = (NettyTransportClient) o;

        return helpCompare.equals(that.helpCompare);
    }

    @Override
    public int hashCode() {
        return helpCompare.hashCode();
    }
}
